package io.github.fb513.api.queue;

import com.fasterxml.jackson.databind.JsonNode;
import com.rabbitmq.client.Channel;
import io.github.fb513.api.facade.LoginFacade;
import io.github.fb513.api.facade.MessageFacade;
import io.github.fb513.common.enums.CommandEnum;
import io.github.fb513.common.model.*;
import io.github.fb513.common.util.JsonParser;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

import static io.github.fb513.common.constant.RabbitConstants.im_to_web_message;

/**
 * @author fengbo
 * @since 2023/12/25 16:21
 */
@Slf4j
@Component
public class MessageReceiver {

    @Autowired
    private ThreadPoolExecutor executor;
    @Autowired
    private JsonParser jsonParser;
    @Autowired
    private LoginFacade loginFacade;
    @Autowired
    private MessageFacade messageFacade;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = im_to_web_message, durable = "true"),
            exchange = @Exchange(value = im_to_web_message, durable = "true"),
            key = im_to_web_message
        ),
            ackMode = "AUTO"
    )
    public void onChatMessage(@Payload Message message,
                              @Headers Map<String,Object> headers,
                              Channel channel) {
        try {
            String msg = new String(message.getBody());
            JsonNode messageInfo = jsonParser.parseJsonToJsonNode(msg);
            MessageInfo.Header header = jsonParser.convertToPojo(messageInfo.get("header"), MessageInfo.Header.class);
            int command = header.getCommand();
            JsonNode content = messageInfo.get("content");
            if (CommandEnum.LOGIN.getCommand() == command) {
                BaseData<?> baseData = jsonParser.convertToPojo(content, BaseData.class, LoginData.class);
                executor.execute(() -> loginFacade.imLogin(baseData));
            } else if (CommandEnum.OFFLINE.getCommand() == command) {
                BaseData<?> baseData = jsonParser.convertToPojo(content, BaseData.class);
                executor.execute(() -> loginFacade.imLogout(baseData));
            } else if (CommandEnum.MESSAGE.getCommand() == command) {
                executor.execute(() -> messageFacade.sendMsg(jsonParser.convertToPojo(content, BaseData.class, MessageData.class)));
            } else if (CommandEnum.CLIENT_MESSAGE_ACK.getCommand() == command) {
                executor.execute(() -> messageFacade.receiveMsg(jsonParser.convertToPojo(content, BaseData.class, MessageRespData.class)));
            } else if (CommandEnum.CLIENT_MESSAGE_READ.getCommand() == command) {
                executor.execute(() -> messageFacade.readMsg(jsonParser.convertToPojo(content, BaseData.class, MessageRespData.class)));
            }
        } catch (Exception e) {
            log.error("消息处理异常", e);
        }
    }
}
